package com.atguigu.fink.chapter03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/11/28 09:06
 */
public class TopN {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        // 1. 建立动态表与数据源关联
        tEnv.executeSql("create table ub(" +
                            " user_id bigint, " +
                            " item_id bigint, " +
                            " category_id int, " +
                            " behavior string, " +
                            " ts bigint, " +
                            " et as to_timestamp_ltz(ts, 0)," +
                            " watermark for et as et - interval '3' second " +
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/UserBehavior.csv', " +
                            " 'format' = 'csv' " +
                            ")");
        
        // 2. 统计每个商品在每个窗口的点击量 (过滤出 pv)
        Table t1 = tEnv
            .sqlQuery("select " +
                          " item_id,  " +
                          " window_start, " +
                          " count(*) ct " +
                          "from table( tumble( table ub, descriptor(et), interval '2' hour) )" +
                          "where behavior='pv' " +
                          "group by item_id, window_start, window_end");
        tEnv.createTemporaryView("t1", t1);
        // 3. 给每个商品的安装点击量降序排列 over 窗口,
        // rank dense_rank row_number(只支持)
        Table t2 = tEnv.sqlQuery("select" +
                                     " item_id, window_start, ct, " +
                                     " row_number() over(partition by window_start order by ct desc) rn " +
                                     "from t1");
        tEnv.createTemporaryView("t2", t2);
        // 4. 过滤 名次<=N
        Table result = tEnv.sqlQuery("select " +
                                         " window_start w_end, " +
                                         " item_id, " +
                                         " ct item_count," +
                                         " rn rk " +
                                         "from t2 " +
                                         "where rn<=3");
        // 5. 结果写出到到 Mysql 中
        tEnv.executeSql("CREATE TABLE `hot_item` ( " +
                            "  `w_end` timestamp , " +
                            "  `item_id` bigint, " +
                            "  `item_count` bigint, " +
                            "  `rk` bigint, " +
                            "  PRIMARY KEY (`w_end`,`rk`) not enforced " +
                            ")with(" +
                            " 'connector' = 'jdbc', " +
                            " 'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false', " +
                            " 'table-name' = 'hot_item', " +
                            " 'username' = 'root', " +
                            " 'password' = 'aaaaaa' " +
                            ")");
    
        result.executeInsert("hot_item");
        
        
    }
}
/*
两次 key:
keyBy 1: 统计每个商品在每个窗口的点击量
keyBy 2: 按照窗口的关闭时间分组, 去 topN
 */